package com.mqtt.service.impl;

import com.alibaba.fastjson.JSON;
import com.mqtt.config.MqttConfiguration;
import com.mqtt.handler.ProtocolHandler;
import com.mqtt.model.DeviceInfo;
import com.mqtt.model.MqttMessage;
import com.mqtt.service.MqttService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * MQTT服务实现
 */
@Slf4j
@Service
public class MqttServiceImpl implements MqttService {
    
    @Autowired
    private ProtocolHandler protocolHandler;
    
    @Autowired
    private DeviceSessionService deviceSessionService;
    
    @Autowired
    private MqttConfiguration mqttConfiguration;
    
    @Autowired
    private MqttPahoMessageDrivenChannelAdapter mqttInboundAdapter;
    
    @Override
    public boolean sendToDevice(String deviceId, String message) {
        String topic = "device/" + deviceId + "/command";
        return sendToTopic(topic, message);
    }
    
    @Override
    public boolean sendToTopic(String topic, String message) {
        return sendToTopic(topic, message, 1, false);
    }
    
    @Override
    public boolean sendToTopic(String topic, String message, int qos, boolean retained) {
        try {
            protocolHandler.sendMessage(topic, message, qos, retained);
            log.debug("Message sent to topic: {}", topic);
            return true;
        } catch (Exception e) {
            log.error("Failed to send message to topic: {}", topic, e);
            return false;
        }
    }
    
    @Override
    public boolean sendCommand(String deviceId, String command, Map<String, Object> params) {
        try {
            // 构建命令消息
            Map<String, Object> commandMessage = new HashMap<>();
            commandMessage.put("commandId", UUID.randomUUID().toString());
            commandMessage.put("command", command);
            commandMessage.put("params", params);
            commandMessage.put("timestamp", System.currentTimeMillis());
            
            String message = JSON.toJSONString(commandMessage);
            String topic = "device/" + deviceId + "/command";
            
            // 发送命令
            boolean sent = sendToTopic(topic, message, 2, false);
            
            if (sent) {
                log.info("Command sent to device {}: {}", deviceId, command);
                
                // 记录命令发送历史
                recordCommandHistory(deviceId, commandMessage);
            }
            
            return sent;
            
        } catch (Exception e) {
            log.error("Failed to send command to device: {}", deviceId, e);
            return false;
        }
    }
    
    @Override
    public void batchSend(List<MqttMessage> messages) {
        if (messages == null || messages.isEmpty()) {
            return;
        }
        
        log.info("Batch sending {} messages", messages.size());
        
        // 使用异步发送提高效率
        List<CompletableFuture<Boolean>> futures = new ArrayList<>();
        
        for (MqttMessage message : messages) {
            CompletableFuture<Boolean> future = CompletableFuture.supplyAsync(() -> 
                sendToTopic(message.getTopic(), message.getPayload(), 
                           message.getQos() != null ? message.getQos() : 1, false)
            );
            futures.add(future);
        }
        
        // 等待所有消息发送完成
        try {
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                    .get(30, TimeUnit.SECONDS);
            
            long successCount = futures.stream()
                    .filter(f -> {
                        try {
                            return f.get();
                        } catch (Exception e) {
                            return false;
                        }
                    })
                    .count();
            
            log.info("Batch send completed: {}/{} messages sent successfully", 
                    successCount, messages.size());
            
        } catch (Exception e) {
            log.error("Error during batch send", e);
        }
    }
    
    @Override
    public void subscribe(String topic) {
        subscribe(topic, 1);
    }
    
    @Override
    public void subscribe(String topic, int qos) {
        try {
            mqttConfiguration.addTopicSubscription(topic);
            log.info("Subscribed to topic: {} with QoS: {}", topic, qos);
        } catch (Exception e) {
            log.error("Failed to subscribe to topic: {}", topic, e);
        }
    }
    
    @Override
    public void unsubscribe(String topic) {
        try {
            mqttConfiguration.removeTopicSubscription(topic);
            log.info("Unsubscribed from topic: {}", topic);
        } catch (Exception e) {
            log.error("Failed to unsubscribe from topic: {}", topic, e);
        }
    }
    
    @Override
    public List<DeviceInfo> getOnlineDevices() {
        return deviceSessionService.getOnlineDevices();
    }
    
    @Override
    public boolean isDeviceOnline(String deviceId) {
        return deviceSessionService.isDeviceOnline(deviceId);
    }
    
    @Override
    public DeviceInfo getDeviceInfo(String deviceId) {
        return deviceSessionService.getDeviceInfo(deviceId);
    }
    
    @Override
    public void disconnectDevice(String deviceId) {
        try {
            // 发送断开连接命令
            Map<String, Object> disconnectCmd = new HashMap<>();
            disconnectCmd.put("action", "disconnect");
            disconnectCmd.put("reason", "server_initiated");
            
            sendCommand(deviceId, "disconnect", disconnectCmd);
            
            // 清除设备会话
            deviceSessionService.removeSession(deviceId);
            
            log.info("Device {} disconnected", deviceId);
            
        } catch (Exception e) {
            log.error("Failed to disconnect device: {}", deviceId, e);
        }
    }
    
    @Override
    public boolean isConnected() {
        try {
            return mqttInboundAdapter != null && mqttInboundAdapter.isRunning();
        } catch (Exception e) {
            log.error("Error checking MQTT connection status", e);
            return false;
        }
    }
    
    @Override
    public void reconnect() {
        try {
            log.info("Attempting to reconnect MQTT...");
            
            // 停止当前连接
            if (mqttInboundAdapter != null && mqttInboundAdapter.isRunning()) {
                mqttInboundAdapter.stop();
            }
            
            // 等待一段时间
            Thread.sleep(2000);
            
            // 重新启动连接
            if (mqttInboundAdapter != null) {
                mqttInboundAdapter.start();
            }
            
            log.info("MQTT reconnected successfully");
            
        } catch (Exception e) {
            log.error("Failed to reconnect MQTT", e);
        }
    }
    
    /**
     * 记录命令历史
     */
    private void recordCommandHistory(String deviceId, Map<String, Object> command) {
        // 实现命令历史记录逻辑
        // 可以保存到Redis或数据库
        log.debug("Recording command history for device: {}", deviceId);
    }
}